package com.gxd.esimport;


import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

public class inputThread implements  Runnable{

    private final Table table;
    private LinkedBlockingQueue<Map<String, Object>> queue;


    public  inputThread(Table table, LinkedBlockingQueue<Map<String, Object>> queue) {
        this.table = table;
        this.queue = queue;
    }

    @Override
    public void run() {

        pipe.atomic.incrementAndGet();
        System.out.println("ThreadName:"+Thread.currentThread().getName()+":"+pipe.atomic.get());


        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        try {

            JdbcClientHelper.getConnection();
            preparedStatement = JdbcClientHelper.conn.prepareStatement(this.table.query, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            preparedStatement.setFetchSize(2000);
            preparedStatement.setFetchDirection(ResultSet.FETCH_REVERSE);
            resultSet = preparedStatement.executeQuery();
            System.out.println("同步语句："+this.table.query);

            // 获取列名
            List<String> columnNameList = new ArrayList<String>();
            ResultSetMetaData rsmd = resultSet.getMetaData();
            int count = rsmd.getColumnCount();
            for (int i = 1; i <= count; i++) {
                columnNameList.add(rsmd.getColumnName(i));
            }


            // 逐行读取
            while (resultSet.next()) {

                Map<String, Object> event = new HashMap<String, Object>();
                Map<String, Object> rowMap = new HashMap<String, Object>();
                for (String columnName : columnNameList) {
                    rowMap.put(columnName, resultSet.getObject(columnName));
                }
                event.put("rowMap", rowMap);
                event.put("index", this.table.index);
                System.out.println(Thread.currentThread().getName()+"入PIPE数据："+String.valueOf(((Map<String, Object>)event.get("rowMap")).get("CITY_CODE"))+"++++"+(String)event.get("index")+"++++"+(String)event.get("index")+"++++"+String.valueOf(((Map<String, Object>)event.get("rowMap")).get("ID")));
                //pipe.simpleArrayConcurrentQueue.put(event);
                this.queue.put(event);
            }

            Thread.currentThread().stop();


        }catch (SQLSyntaxErrorException e){
            //e.printStackTrace();
            System.out.println("线程关闭！！！查血SQl：" + this.table.query);

        }catch (Exception e) {

            System.out.println("read failed:"+this.table.query);
            e.printStackTrace();
            Thread.currentThread().stop();
        } finally {
            //JdbcClientHelper.releaseConnection(JdbcClientHelper.conn, preparedStatement, resultSet);
            try {
                preparedStatement.close();
                Thread.currentThread().stop();
            } catch (SQLException e) {
                e.printStackTrace();
            }

        }

    }

}
